package com.learn.kafka.stream;

import com.learn.kafka.stream.config.KafkaStreamProperties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.TimeWindows;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * Kafka Stream 流处理示例（继承通用配置父类）
 * Kafka Stream处理应用：
 * 1. 统计单词出现次数
 * 2. 提取长度大于6的单词
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/9 16:39
 * @updateUser: MI
 * @updateTime: 2025/8/9 16:39
 * @updateRemark: 修改内容
 * @version: 1.0
 */
public class KafkaStreamProcessor extends KafkaStreamProperties {


    public static void main(String[] args) {
        new KafkaStreamProcessor().startStreamProcessing();
    }

    // 构造方法：初始化当前流应用的特殊配置
    public KafkaStreamProcessor() {
        super(); // 调用父类构造方法，加载通用配置
        initStreamConfig();
    }

    // 初始化当前应用的特殊配置（覆盖父类默认值）
    private void initStreamConfig() {
        // 设置应用ID（覆盖父类默认值）
        setConfig(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-stream-app");
        // 如需修改集群地址，可在此处覆盖
        // setConfig(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.44.160:9092");
    }

    // 启动流处理逻辑
    private void startStreamProcessing() {
        // 1. 构建流处理拓扑
        StreamsBuilder builder = new StreamsBuilder();

        // 从输入主题"text-input"读取数据
        KStream<String, String> textLines = builder.stream("text-input");

        // 2. 实现单词计数逻辑
        KTable<String, Long> wordCounts = textLines
                // 将值转换为小写
                .mapValues(textLine -> textLine.toLowerCase(Locale.ROOT))
                // 按非单词字符拆分字符串为单词数组
                .flatMapValues(textLine -> Arrays.asList(textLine.split("\\W+")))
                // 过滤空字符串
                .filter((key, value) -> !value.isEmpty())
                // 以单词为键，值为1（用于计数）
                .selectKey((key, value) -> value)
                // 按单词分组
                .groupByKey()
                // 窗口计数（5分钟窗口）
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))).count()
                // 转换窗口化结果为普通KTable
                .toStream().selectKey((windowedKey, count) -> windowedKey.key())
                .groupByKey().reduce((aggValue, newValue) -> aggValue + newValue);

        // 将单词计数结果输出到"tream-word-count"主题
        wordCounts.toStream().to("stream-word-count", Produced.with(Serdes.String(), Serdes.Long()));

        // 3. 实现数据过滤逻辑：筛选长度大于5的单词
//        KStream<String, String> longWords = textLines.flatMapValues(textLine -> Arrays.asList(textLine.split("\\W+")))
//                .filter((key, value) -> value.length() > 5).selectKey((key, value) -> value);
        // 分支2：提取长单词（长度>6）
        // KafkaStreamProcessor 中长单词处理逻辑
        KStream<String, String> longWords = textLines
                // 拆分并过滤空单词
                .flatMapValues(text -> {
                    List<String> words = Arrays.asList(text.split("\\W+"));
                    // 过滤空字符串和null
                    return words.stream()
                            .filter(word -> word != null && !word.isEmpty())
                            .collect(Collectors.toList());
                })
                // 筛选长度≥6的单词（确保word非空）
                .filter((key, word) -> word.length() >= 6)
                // 明确将key设置为单词
                .selectKey((key, word) -> word);

        // 将长单词结果输出到"long-words-output"主题
//        longWords.to("long-words-output");
        longWords.to("stream-long-words",
                Produced.with(Serdes.String(), Serdes.String()));  // 显式指定key使用String序列化器

        // 4. 从父类获取KafkaStreams实例并启动
        final KafkaStreams streams = createKafkaStreams(builder.build());
        final CountDownLatch latch = new CountDownLatch(1);

        // 注册关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

        try {
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
}
